package com.atguigu.realtime.streaming.apps

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.realtime.constants.{PrefixConstant, TopicConstant}
import com.atguigu.realtime.streaming.apps.GMVApp.{appName, batchDuration, context}
import com.atguigu.realtime.streaming.utils.{MyKafkaUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * Created by Smexy on 2022/5/28
 *
 *  如何在redis中存储用户的信息?
 *       K: String。 体现粒度和唯一性
 *       v:
 *          单值: string, hash
 *          集合: set,list,zset
 *
 *    ---------------------
 *      所有的用户存储一个Key
 *      K: user
 *      V: Set
 *    ----------------------
 *      一个用户存储一个Key
 *      K: user:id
 *      V: String
 *
 *    ------------------
 *      场景:   order_info 关联 order_detail
 *                有user_id，使用user_id去查询用户的其他信息。
 *                一个用户存储一个K-V更合适，节省查询效率。
 *
 *
 *    --------------------------
 *      RDD的初始分区数取决于 Kafka主题的分区数
 *
 *
 */
object UserInfoApp extends BaseApp {
  override var appName: String = "UserInfoApp"
  override var batchDuration: Int = 10
  override var groupId: String = "saledetail_user"
  override var topic: String = TopicConstant.USER_INFO

  def main(args: Array[String]): Unit = {

    context = new StreamingContext("local[*]", appName, Seconds(batchDuration))

    runApp{

      val ds: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Array(topic), context, groupId)

      ds.foreachRDD(rdd => {

        if (!rdd.isEmpty()){

          val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

          //以分区为单位写入
          rdd.foreachPartition(partition=>{

            val jedis: Jedis = RedisUtil.getJedisClient()

            //打印统计,调用size后，迭代器已经迭代到最后了
            // 如果对一个迭代器迭代多次，只能把迭代器转为集合才可以
            val list: List[ConsumerRecord[String, String]] = partition.toList

            println("即将写入:"+list.size)

            list.foreach(record => {

              val map: JSONObject = JSON.parseObject(record.value())

              val key = PrefixConstant.user_info_redis_preffix + map.getString("id")

              jedis.set(key,record.value())

            })

            jedis.close()


          })

          //提交偏移量
          ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

        }
      })
    }
  }
}
